package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/5/24 9:18
 */
public class Dwd_10_DwdTradeRefundPaySucApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_10_DwdTradeRefundPaySucApp().init(2010, 2, "Dwd_10_DwdTradeRefundPaySucApp");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // ttl设置多少?
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        
        // 1. 要过滤退单数据, 需要先读取ods_db
        readOdsDb(tEnv, "Dwd_09_DwdTradeOrderRefundApp");
        // 2. 读取字典表
        readBaseDic(tEnv);
        // 3. 过滤退单数据
        Table orderRefundInfo = tEnv.sqlQuery("select " +
                                                  "data['id'] id, " +
                                                  "data['user_id'] user_id, " +
                                                  "data['order_id'] order_id, " +
                                                  "data['sku_id'] sku_id, " +
                                                  "data['refund_type'] refund_type, " +
                                                  "data['refund_num'] refund_num, " +
                                                  "data['refund_amount'] refund_amount, " +
                                                  "data['refund_reason_type'] refund_reason_type, " +
                                                  "data['refund_reason_txt'] refund_reason_txt, " +
                                                  "data['create_time'] create_time, " +
                                                  "pt, " +
                                                  "ts " +
                                                  "from ods_db " +
                                                  "where `table`='order_refund_info' " +
                                                  "and `type`='insert'");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);
        
        // 4. 过滤oder_info: 退款成功
        Table orderInfo = tEnv.sqlQuery("select " +
                                            "data['id'] id, " +
                                            "data['user_id'] user_id, " +
                                            "data['province_id'] province_id, " +
                                            "`old` " +
                                            "from ods_db " +
                                            "where `table`='order_info' " +
                                            "and `type`='update' " +
                                            "and `old`['order_status'] is not null " +
                                            "and `data`['order_status']='1006'");
        
        tEnv.createTemporaryView("order_info", orderInfo);
        
        // 5. 退款表`
        Table refundPayment = tEnv.sqlQuery("select " +
                                                "data['id'] id, " +
                                                "data['order_id'] order_id, " +
                                                "data['sku_id'] sku_id, " +
                                                "data['payment_type'] payment_type, " +
                                                "data['callback_time'] callback_time, " +
                                                "data['total_amount'] total_amount, " +
                                                "pt, " +
                                                "ts " +
                                                "from ods_db " +
                                                "where `table`='refund_payment' "
                                            //                          +
                                            //                          "and `type`='update' " +
                                            //                          "and `old`['refund_status'] is not null " +
                                            //                          "and `refund_status`='0705' "  // 退款完成
        );
        tEnv.createTemporaryView("refund_payment", refundPayment);
        
        // 6. join
        Table resultTable = tEnv.sqlQuery("select " +
                                        "rp.id, " +
                                        "oi.user_id, " +
                                        "rp.order_id, " +
                                        "rp.sku_id, " +
                                        "oi.province_id, " +
                                        "rp.payment_type, " +
                                        "dic.dic_name payment_type_name, " +
                                        "date_format(rp.callback_time,'yyyy-MM-dd') date_id, " +
                                        "rp.callback_time, " +
                                        "ri.refund_num, " +
                                        "rp.total_amount, " +
                                        "rp.ts, " +
                                        "current_row_timestamp() row_op_ts " +
                                        "from refund_payment rp " +
                                        "join order_info oi on rp.order_id=oi.id " +
                                        "join order_refund_info ri on rp.order_id=ri.order_id and rp.sku_id=ri.sku_id " +
                                        "join base_dic for system_time as of rp.pt dic on rp.payment_type=dic.dic_code ");
        
        
        tEnv.executeSql("create table dwd_trade_refund_pay_suc( " +
                                "id string, " +
                                "user_id string, " +
                                "order_id string, " +
                                "sku_id string, " +
                                "province_id string, " +
                                "payment_type_code string, " +
                                "payment_type_name string, " +
                                "date_id string, " +
                                "callback_time string, " +
                                "refund_num string, " +
                                "refund_amount string, " +
                                "ts string, " +
                                "row_op_ts timestamp_ltz(3) " +
                                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_TRADE_REFUND_PAY_SUC));
    
        resultTable.executeInsert("dwd_trade_refund_pay_suc");
    
    
    }
}
